package pers.twf.twfRpc.registry;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.cron.CronUtil;
import cn.hutool.cron.task.Task;
import cn.hutool.json.JSONUtil;
import io.lettuce.core.KeyScanCursor;
import io.lettuce.core.RedisClient;
import io.lettuce.core.RedisURI;
import io.lettuce.core.ScanArgs;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.api.sync.RedisCommands;
import io.lettuce.core.codec.RedisCodec;
import io.lettuce.core.event.EventBus;
import io.lettuce.core.pubsub.RedisPubSubAdapter;
import io.lettuce.core.pubsub.RedisPubSubListener;
import io.lettuce.core.pubsub.StatefulRedisPubSubConnection;
import io.lettuce.core.pubsub.api.sync.RedisPubSubCommands;
import pers.twf.twfRpc.config.RegistryConfig;
import pers.twf.twfRpc.model.ServiceMetaInfo;

import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.List;
import java.util.Optional;
import java.util.stream.Collectors;

public class RedisRegistry implements Registry {
    /**
     * Redis客户端实例，用于与Redis服务器进行交互
     */
    private RedisClient redisClient;

    /**
     * 状态化的Redis连接，用于管理与Redis服务器的连接状态
     */
    private StatefulRedisConnection<String, ServiceMetaInfo> connection;

    /**
     * 同步命令的Redis命令执行器，用于执行各种Redis命令
     */
    private RedisCommands<String, ServiceMetaInfo> sysncCommands;

    private StatefulRedisPubSubConnection<String, String> pubSubConnection;


    private RedisPubSubCommands<String, String> pubSubCommands;

    /**
     * 根节点
     */
    private static final String REDIS_ROOT_PATH = "rpc:redis";

    @Override
    public void heartBeat() {
//        每10秒续签一次
        CronUtil.schedule("*/10 * * * * *", new Task() {
            @Override
            public void execute() {
                for (String registerKey : localRegisterNodeKeySet) {
                    try {
//                       获取节点
                        Optional<ServiceMetaInfo> serviceMetaInfo = Optional.ofNullable(sysncCommands.get(registerKey));
                        if (serviceMetaInfo.isEmpty()) {
                            continue;
                        }
                        register(serviceMetaInfo.get());
                    } catch (Exception e) {
                        throw new RuntimeException(registerKey + "续签失败", e);
                    }

                }
            }
        });
//         支持秒级定时任务
        CronUtil.setMatchSecond(true);
        CronUtil.start();
    }


    @Override
    public void watch(String serviceKey){
        //todo 未实现监听
    }

    @Override
    public void init(RegistryConfig registryConfig) {
        RedisURI redisUri = RedisURI.create(registryConfig.getAddress());

        // 如果需要设置超时时间和认证信息，可以继续使用 builder 模式进行修改
        redisUri.setTimeout(Duration.ofMillis(registryConfig.getTimeout()));

        // 创建Redis客户端
        redisClient = RedisClient.create(redisUri);
        // 建立连接,并定义编解码器
        connection = redisClient.connect(new RedisCodec<>() {
            @Override
            public String decodeKey(ByteBuffer byteBuffer) {
                byte[] bytes = new byte[byteBuffer.remaining()];
                byteBuffer.duplicate().get(bytes);
                return new String(bytes, StandardCharsets.UTF_8);
            }

            @Override
            public ServiceMetaInfo decodeValue(ByteBuffer byteBuffer) {
                // 检查是否可以直接访问底层数组
                if (!byteBuffer.hasArray()) {
                    // 如果不能直接访问底层数组，创建一个新的堆缓冲区并复制数据
                    ByteBuffer heapBuffer = ByteBuffer.allocate(byteBuffer.remaining());
                    heapBuffer.put(byteBuffer);
                    heapBuffer.flip();
                    byteBuffer = heapBuffer;
                }

                // 现在可以安全地调用 array() 方法
                byte[] bytes = byteBuffer.array();
                return JSONUtil.toBean(new String(bytes, StandardCharsets.UTF_8), ServiceMetaInfo.class);
            }

            @Override
            public ByteBuffer encodeKey(String key) {
                return ByteBuffer.wrap(key.getBytes(StandardCharsets.UTF_8));
            }

            @Override
            public ByteBuffer encodeValue(ServiceMetaInfo serviceMetaInfo) {
                return ByteBuffer.wrap(JSONUtil.toJsonStr(serviceMetaInfo).getBytes(StandardCharsets.UTF_8));
            }
        });
        // 获取同步命令执行器
        sysncCommands = connection.sync();
        heartBeat();
    }

    @Override
    public void register(ServiceMetaInfo serviceMetaInfo) throws Exception {
//        设置要存储的键值对
        String registerKey = REDIS_ROOT_PATH + ":" + serviceMetaInfo.getServiceNodeRedisKey();
//        关联键值对和租约，设置过期时间
        sysncCommands.set(registerKey, serviceMetaInfo);
        sysncCommands.expire(registerKey, 30);

        localRegisterNodeKeySet.add(registerKey);
    }

    @Override
    public void unRegister(ServiceMetaInfo serviceMetaInfo) {
        String registerKey = REDIS_ROOT_PATH + ":" + serviceMetaInfo.getServiceNodeRedisKey();
        sysncCommands.del(registerKey);
        localRegisterNodeKeySet.remove(registerKey);
    }

    @Override
    public List<ServiceMetaInfo> serviceDiscovery(String serviceKey) {
//        优先从缓存获取服务
        List<ServiceMetaInfo> cacheServiceMetaInfos = registryServiceCache.readCache(serviceKey);
        if (CollUtil.isNotEmpty(cacheServiceMetaInfos)) {
            return cacheServiceMetaInfos;
        }
        // 前缀搜索，结尾一定要加 :
        String searchPrefix = REDIS_ROOT_PATH + ":" + serviceKey + ":";
        try {
            // 前缀查询
            ScanArgs scanArgs = ScanArgs.Builder.matches(searchPrefix+"*");
            KeyScanCursor<String> keys = sysncCommands.scan(scanArgs);
            // 解析服务信息
            List<ServiceMetaInfo> serviceMetaInfoList = keys.getKeys().stream()
                    .map(keyValue -> {
                        return sysncCommands.get(keyValue);
                    })
                    .collect(Collectors.toList());
//            写入服务缓存
            registryServiceCache.writeCache(serviceKey, serviceMetaInfoList);
            return serviceMetaInfoList;
        } catch (Exception e) {
            throw new RuntimeException("获取服务列表失败", e);
        }
    }

    @Override
    public void destroy() {
        System.out.println("当前节点下线");
        for (String key : localRegisterNodeKeySet) {
            try {
                sysncCommands.del(key);
            } catch (Exception e) {
                throw new RuntimeException(key+"节点下线失败");
            }
        }
        // 释放资源
        if (connection != null) {
            connection.close();
        }
        if (redisClient != null) {
            redisClient.shutdown();
        }
    }
}
